package com.bwie.list.consumser;

import com.bwie.common.domain.Data3;
import com.bwie.list.service.TestService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class ConsumserData3 {
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private TestService service;

    @KafkaListener(topics = "topic3",containerFactory = "kafkaListenerContainerFactory",errorHandler = "myKafkaListenerErrorHandler")
    public void ConsumserData3(ConsumerRecord<Object,Object> consumerRecord, Acknowledgment acknowledgment){
        String key = (String)consumerRecord.key();
        Long add = redisTemplate.opsForSet().add(key);
        if(add==1){
            Data3 data3 = (Data3) consumerRecord.value();
            service.addData3(data3);
            acknowledgment.acknowledge();
        }

    }

}
